package vip.ifmm.lehouse.mq.service.api.impl;

import com.alibaba.dubbo.config.annotation.Service;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Value;
import vip.ifmm.lehouse.common.dto.ServerResponse;
import vip.ifmm.lehouse.common.enums.status.MqStatus;
import vip.ifmm.lehouse.mq.service.api.RocketMqService;
import vip.ifmm.lehouse.mq.service.utils.RandomUtil;
import javax.annotation.PostConstruct;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.List;


/**
 * @author Red
 * email: 1318944013@qq.com
 * date: 2019/6/17 22:05
 */
@Slf4j
@Service(owner = "red",
         mock = "vip.ifmm.lehouse.mq.service.api.mock.RocketMqServiceMock")
public class RocketMqServiceImpl implements RocketMqService {

    // 读取配置文件中的RocketMQ服务所在主机端口号
    @Value("${namesrv-addr}")
    private  String NAMESER_ADDR = null;
    // 生产者个数
    @Value("${produces-size}")
    private int PRODUCERS_SIZE = 0;

    // 默认消息key
    private final String DEFAULT_MSG_KEY = "default-producer-key";
    // 默认生产者组
    private final String DEFAULT_PRODUCER_GROUP = "default-producer-group";
    // 存放生产者的集合
    public static List<DefaultMQProducer> producerList = null;

    @PostConstruct
    public void init() {
        // 创建生产者们
        createSizeProducer();
        log.info("生产者初始化完成, 个数为: " + PRODUCERS_SIZE);

    }

    // 默认生产方法
    @Override
    public ServerResponse<SendResult> send(String topic, String tags, String msg) {

        // 创建消息
        Message message = getMessage(topic, tags, msg);
        SendResult sendResult = null;
        try {
            // 发送消息, 超时时间为5秒
            sendResult = getProducerRandom().send(message, 5000);
        } catch (MQClientException | MQBrokerException | RemotingException | InterruptedException e) {
            e.printStackTrace();
        }

        if(sendResult == null || sendResult.getSendStatus() != SendStatus.SEND_OK) {
            return ServerResponse.response(MqStatus.MSG_SEND_FAILURE);
        } else {
            return ServerResponse.response(MqStatus.MSG_SEND_SUCCESS,sendResult);
        }
    }

    // 顺序生产方法
    @Override
    public ServerResponse<SendResult> sendOrderly(String topic, String tags,
                                  String msg, Integer queueIndex) {
        // 创建消息
        Message message = getMessage(topic, tags, msg);
        SendResult sendResult = null;
        try {
            // 往指定队列里发送消息
            sendResult = getProducerRandom().send(message, new MessageQueueSelector() {
                @Override
                public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
                    // 获得下标为0的队列
                    Integer index = (Integer) o;
                    return list.get(index);
                }
            }, queueIndex, 5000);// 指定队列下标
        } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
            e.printStackTrace();
        }
        // 是否发送成功
        if(sendResult == null || sendResult.getSendStatus() != SendStatus.SEND_OK) {
            return ServerResponse.response(MqStatus.MSG_SEND_FAILURE);
        } else {
            return ServerResponse.response(MqStatus.MSG_SEND_SUCCESS, sendResult);
        }
    }

    // 获得默认消费者
    private DefaultMQProducer getProducerRandom() {
        return producerList.get(RandomUtil.randomNumber(PRODUCERS_SIZE));
    }

    // 组装消息
    private Message getMessage(String topic, String tags, String msg) {
        Message message = null;
        try {
            // 创建消息 参数：主题 标签(过滤用) 消息的key 消息体
            message = new Message(topic, tags, DEFAULT_MSG_KEY,
                    msg.getBytes(RemotingHelper.DEFAULT_CHARSET));
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        }
        return message;
    }

    // 设置每个producer的属性
    private void initProducer(DefaultMQProducer producer) {
        // 设置RocketMQ服务所在主机端口号
        producer.setNamesrvAddr(NAMESER_ADDR);
        // 开启生产者
        try {
            producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

    // 创建指定个数的消费者，并添加到list中存储
    private void createSizeProducer() {
        // 初始化集合大小
        producerList = new ArrayList<>(PRODUCERS_SIZE);
        for(int i = 0; i < PRODUCERS_SIZE; i++) {
            DefaultMQProducer producer = new DefaultMQProducer(DEFAULT_PRODUCER_GROUP + i);
            // 初始化生产者属性
            this.initProducer(producer);
            producerList.add(producer);
        }
    }

}
